package com.atguigu.gmall.realtime.app.func;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.bean.TableProcess;
import com.atguigu.gmall.realtime.common.GmallConfig;
import com.atguigu.gmall.realtime.util.DruidDSUtil;
import com.atguigu.gmall.realtime.util.PhoenixUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * @author Felix
 * @date 2022/7/27
 * 对维度数据进行处理
 */
public class TableProcessFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {

    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;

    private DruidDataSource dataSource;

    public TableProcessFunction(MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        //获取连接池对象
        dataSource = DruidDSUtil.createDataSource();
    }

    // {"database":"gmall0212","table":"comment_info","type":"insert","ts":1658816554,
    // "data":{"id":1551815221317578756,"user_id":14,"sku_id":29,"spu_id":10,"order_id":11,"appraise":"1204"}}
    //处理主流数据
    @Override
    public void processElement(JSONObject jsonObj, ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        //获取当前操作的业务数据库表的表名
        String tableName = jsonObj.getString("table");
        //获取广播状态
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        //根据业务数据库表的表名 到广播状态中获取对应的配置信息
        TableProcess tableProcess = broadcastState.get(tableName);
        //说明当前处理的是维度表
        if(tableProcess != null){
            //{"tm_name":"华为111","logo_url":"/static/default.jpg","id":3}
            JSONObject dataJsonObj = jsonObj.getJSONObject("data");

            //在向下游传递数据之前，将不需要传递的属性过滤掉
            String sinkColumns = tableProcess.getSinkColumns();
            filterColumn(dataJsonObj, sinkColumns);

            //根据配置信息获取输出到phoenix的目的地表名
            String sinkTable = tableProcess.getSinkTable();
            //在向下游传递数据前，将输出的目的地携带上
            dataJsonObj.put("sink_table",sinkTable);

            String type = jsonObj.getString("type");
            dataJsonObj.put("type",type);

            //将维度表继续向下游传递
            out.collect(dataJsonObj);
        }
    }

    //过滤不需要传递的属性
    // dataJsonObj:{"tm_name":"华为111","logo_url":"/static/default.jpg","id":3}
    // sinkColumns: id,tm_name
    private void filterColumn(JSONObject dataJsonObj, String sinkColumns) {
        String[] fieldArr = sinkColumns.split(",");
        List<String> fieldList = Arrays.asList(fieldArr);
        Set<Map.Entry<String, Object>> entrySet = dataJsonObj.entrySet();

        entrySet.removeIf(entry->!fieldList.contains(entry.getKey()));

    }

    /**
     * read:
     * {"op":"r","after":{"sink_table":"dim_base_category2","source_table":"base_category2","sink_pk":"id","sink_columns":"id,name,category1_id"},"source":{"server_id":0,"version":"1.5.4.Final","file":"","connector":"mysql","pos":0,"name":"mysql_binlog_source","row":0,"ts_ms":1658904042400,"snapshot":"false","db":"gmall0212_config","table":"table_process"},"ts_ms":1658904042400}
     * insert
     * {"op":"c","after":{"sink_table":"dim_test","source_table":"tttest","sink_pk":"id","sink_columns":"id,name"},"source":{"server_id":1,"version":"1.5.4.Final","file":"mysql-bin.000013","connector":"mysql","pos":774567,"name":"mysql_binlog_source","row":0,"ts_ms":1658904125000,"snapshot":"false","db":"gmall0212_config","table":"table_process"},"ts_ms":1658904124940}
     * update
     * {"op":"u","before":{"sink_table":"dim_test","source_table":"tttest","sink_pk":"id","sink_columns":"id,name"},"after":{"sink_table":"dim_test","source_table":"tttest","sink_pk":"id","sink_columns":"id,name,age"},"source":{"server_id":1,"version":"1.5.4.Final","file":"mysql-bin.000013","connector":"mysql","pos":774896,"name":"mysql_binlog_source","row":0,"ts_ms":1658904179000,"snapshot":"false","db":"gmall0212_config","table":"table_process"},"ts_ms":1658904179122}
     * delete
     * {"op":"d","before":{"sink_table":"dim_test","source_table":"test","sink_pk":"id","sink_columns":"id,name,age"},"source":{"server_id":1,"version":"1.5.4.Final","file":"mysql-bin.000013","connector":"mysql","pos":775630,"name":"mysql_binlog_source","row":0,"ts_ms":1658904292000,"snapshot":"false","db":"gmall0212_config","table":"table_process"},"ts_ms":1658904292598}
     */
    //处理广播流数据
    @Override
    public void processBroadcastElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {
        //从中获取FlinkCDC从配置表中读取到的配置信息，为了处理方便  将其由jsonStr->jsonObj
        JSONObject jsonObj = JSON.parseObject(jsonStr);
        // System.out.println(jsonObj);
        //获取广播状态
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);

        String op = jsonObj.getString("op");
        //判断对配置表的操作类型
        if ("d".equals(op)) {
            //从配置表中删除了一条配置，我们需要将对应的配置从广播状态中也删除掉
            TableProcess beforeTableProcess = jsonObj.getObject("before", TableProcess.class);
            broadcastState.remove(beforeTableProcess.getSourceTable());
        } else {
            //对配置表做了新增、读取或者是修改操作，需要将对应的配置信息放到广播状态中
            TableProcess tableProcess = jsonObj.getObject("after", TableProcess.class);

            //获取业务数据库中维度表表名
            String sourceTable = tableProcess.getSourceTable();
            //获取数仓中维度表表名
            String sinkTable = tableProcess.getSinkTable();
            //获取维度表中字段
            String sinkColumns = tableProcess.getSinkColumns();
            //获取建表主键
            String sinkPk = tableProcess.getSinkPk();
            //获取建表扩展语句
            String sinkExtend = tableProcess.getSinkExtend();

            //提前将phoenix中的维度表创建出来
            checkTable(sinkTable, sinkColumns, sinkPk, sinkExtend);

            //将配置信息放到状态中
            broadcastState.put(sourceTable, tableProcess);
        }

    }

    //提前将维度表创建出来
    private void checkTable(String tableName, String sinkColumns, String pk, String ext) {
        if (pk == null) {
            pk = "id";
        }
        if (ext == null) {
            ext = "";
        }

        //拼接建表语句
        StringBuilder createSql = new StringBuilder("create table if not exists " + GmallConfig.PHOENIX_SCHEMA + "." + tableName + "(");
        String[] columnArr = sinkColumns.split(",");
        for (int i = 0; i < columnArr.length; i++) {
            String columnName = columnArr[i];
            //判断是否为主键
            if(pk.equals(columnName)){
                createSql.append(columnName + " varchar primary key");
            }else{
                createSql.append(columnName + " varchar");
            }
            if(i < columnArr.length - 1){
                createSql.append(",");
            }
        }
        createSql.append(") " + ext);
        System.out.println("在phoenix中建表语句:" + createSql);


        try {
            Connection conn = dataSource.getConnection();
            PhoenixUtil.executeSql(createSql.toString(),conn);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
